﻿/*----------------------------------------------------------------
// Copyright (C) 2016 巨人网络
//
// 模块名：asio_server
// 创建者：Jin Qing (http://blog.csdn.net/jq0123)
// 修改者列表：
// 创建日期：2016.5.14
// 模块描述：Asio服务器
//----------------------------------------------------------------*/

#include "asio_server.h"

#include <boost/asio/steady_timer.hpp>  // for steady_timer

#include "asio_s2s_session.h"  // for CAsioS2sSession
#include "asio_svr_session.h"  // for CAsioSvrSession
#include "client_mailbox.h"  // for CClientMailBox
#include "log.h"
#include "pb/rpc/rpc.pb.h"  // for RpcRequest
#include "rpc/rpc_call_context.h"  // for CRpcCallContext
#include "rpc/rpc_req_handler.h"  // for CRpcReqHandler
#include "server_mailbox.h"  // for CServerMailBox
#include "server_mailbox_mgr.h"  // for GetSvrMailBoxes()
#include "server_type.h"  // for SERVER_MAILBOX_RESERVE_SIZE
#include "timer_queue_root.h"  // for TimerQueueRoot
#include "world.h"  // for GetSvrMbMgr()

const char LOG_NAME[] = "CAsioServer";

using namespace std;

CAsioServer::CAsioServer(uint16_t uSvrId, EServerType eServerType,  CWorld& rWorld) :
	m_uServerId(uSvrId),
	m_eServerType(eServerType),
	m_rWorld(rWorld),
	m_pRpcReqHandler(new CRpcReqHandler(*this))
{
}

CAsioServer::~CAsioServer()
{
}

// Accept coroutine.
void CAsioServer::LoopAccept(unsigned short port,
	boost::asio::yield_context yield)
{
	using boost::asio::ip::tcp;
	tcp::acceptor acceptor(GetIoService(),
		tcp::endpoint(tcp::v4(), port));

	for (;;)
	{
		boost::system::error_code ec;
		std::shared_ptr<CAsioSvrSession> pSession(new CAsioSvrSession(*this));
		acceptor.async_accept(pSession->GetSocket(), yield[ec]);
		if (ec)
			continue;
		pSession->Go();
	}
}

// Timer.
static void LoopTimer(boost::asio::io_service& rIos,
	boost::asio::yield_context yield)
{
	boost::asio::steady_timer timer(rIos);
	TimerQueueRoot& rTimerQueueRoot = TimerQueueRoot::Get();
	for (;;)
	{
		rTimerQueueRoot.Tick();
		timer.expires_from_now(std::chrono::milliseconds(100));
		boost::system::error_code ec;
		timer.async_wait(yield[ec]);
		if (ec)
			LOG_ERROR(Fmt("Async wait error: (%d)%s") % ec.value() % ec.message());
	}
}

// 注册服务。
// sServiceName须包含包名，如"rpc.TestSvc".
// pService允许为空，表示忽略该服务。
void CAsioServer::RegisterService(
	const std::string& sServiceName,
	RpcServiceSptr pService)
{
	m_pRpcReqHandler->RegisterService(sServiceName, pService);
}

void CAsioServer::RegisterService(RpcServiceSptr pService)
{
	m_pRpcReqHandler->RegisterService(pService);
}

void CAsioServer::HandleRpcRequest(uint64_t uRpcCltId,
	const rpc::RpcRequest& req)
{
	m_pRpcReqHandler->HandleRpcRequest(uRpcCltId, req);
}

void CAsioServer::HandleRpcRequest(const CRpcCallContext& ctx,
	const std::string& sService, const std::string& sMethod,
	const std::string& sContent)
{
	m_pRpcReqHandler->HandleRpcRequest(ctx, sService, sMethod, sContent);
}

void CAsioServer::BlockingRun(uint16_t uPort)
{
	if (0 == m_uServerId)
	{
		LOG_FATAL("Server ID is 0!");
		return;
	}
	LOG_INFO_TO("CAsioServer.Service", Fmt("Listen on port %u") % uPort);

	// See: doc/html/boost_asio/example/cpp11/spawn/echo_server.cpp
	try
	{
		using namespace boost::asio;
		spawn(m_ioService, [this, uPort](yield_context yield){
			LoopAccept(uPort, yield);
		});
		spawn(m_ioService, [this](yield_context yield){
			LoopTimer(m_ioService, yield);
		});
		SpawnS2sSessions();  // Connect to other servers.
		DoSpawnPushPluto();  // 仅用于 CDbMgrServer。Todo(Jinq): Del

		m_ioService.run();
	}
	catch (std::exception& e)
	{
		LOG_ERROR("Exception: " << e.what());
	}
}

// 服务器主动关闭一个socket. 以 ClientId 为索引， 即MailBoxId, SessionId.
void CAsioServer::DisconnectClient(uint64_t uCltId)
{
	CAsioSvrSession* pSession = GetSession(uCltId);
	if (pSession)
		pSession->Disconnect();
}

void CAsioServer::InsertSession(CAsioSvrSession& rSession)
{
	m_mapSessions[rSession.GetId()] = &rSession;
}

void CAsioServer::EraseSession(CAsioSvrSession& rSession)
{
	m_mapSessions.erase(rSession.GetId());
}

CClientMailBox* CAsioServer::GetClientMailbox(uint64_t uMbId)
{
	// uMbId == uSessionId
	CAsioSvrSession* pSession = GetSession(uMbId);
	if (pSession)
		return &(pSession->GetCltMailBox());
	return nullptr;
}

void CAsioServer::Shutdown()
{
	m_bShutdown = true;
	LOG_INFO("Shutdown()");
}

void CAsioServer::SpawnS2sSessions()
{
	ServerMailBoxList& mbs = m_rWorld.GetSvrMbMgr().GetSvrMailBoxes();
	m_serverMbs.resize(SERVER_MAILBOX_RESERVE_SIZE);
	assert(!m_serverMbs[0]);  // inited to nullptr

	auto iter = mbs.begin();
	for (; iter != mbs.end(); )
	{
		CServerMailBox* pmb = *iter;
		if (m_uServerId == pmb->GetServerId())
		{
			delete pmb;
			*iter = nullptr;
			iter = mbs.erase(iter);
			continue;
		}

		m_serverMbs[pmb->GetServerId()] = pmb;
		++iter;
		auto pSession = std::make_shared<CAsioS2sSession>(*this, *pmb);
		pSession->Go();
	}
}

CAsioSvrSession* CAsioServer::GetSession(uint64_t uSessionId) const
{
	auto iter = m_mapSessions.find(uSessionId);
	if (iter == m_mapSessions.end())
		return nullptr;
	return (*iter).second;
}
